package top.houry.netty.barrage.rocketmq;

import cn.hutool.json.JSONUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;
import top.houry.netty.barrage.entity.BarrageMsg;
import top.houry.netty.barrage.service.IBarrageMsgService;

import java.util.List;

/**
 *
 * RocketMQ
 *  消费者
 */
@Component
@Slf4j
public class ConsumerService implements ApplicationContextAware {
    final static String nameServerUrl = "127.0.0.1:9876";
    final static String topic = "db";
    final static String ConsumerGROUP = "test-consumer";
    private IBarrageMsgService barrageMsgService;

    @Autowired
    public void setBarrageMsgService(IBarrageMsgService barrageMsgService) {
        this.barrageMsgService = barrageMsgService;
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        initConsumer();
    }

    synchronized private void initConsumer() {
        DefaultMQPushConsumer mqPushConsumer = new DefaultMQPushConsumer(ConsumerGROUP);
        mqPushConsumer.setNamesrvAddr(nameServerUrl);
        // 开启广播模式
        // 广播模式下，发送一条消息后，会被当前被广播的所有Consumer消费，但是后面新加入的Consumer不会消费这条消息
//        mqPushConsumer.setMessageModel(MessageModel.BROADCASTING);
        // *指定的是tag标签，代表全部消息，不进行任何过滤
        try {
            mqPushConsumer.subscribe(topic,"*");
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        // 注册监听器，进行消息消费
        mqPushConsumer.registerMessageListener((MessageListenerConcurrently) (list, consumeConcurrentlyContext) -> {

            for (MessageExt messageExt : list) {
                String message = new String(messageExt.getBody());
                BarrageMsg msg = JSONUtil.toBean(message, BarrageMsg.class);
                barrageMsgService.saveBarrageMsg(msg);
                log.info("收到===================== topic:distributed 的消息【{}】",message);
            }
            log.info("插入数据库成功!");
            // ack确认
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        //启动消费者
        try {
            mqPushConsumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        log.info("test-consumer 消费者已经启动");
    }
}
